package engine.BIEngine0_1;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.transform.OutputKeys;
import javax.xml.transform.Transformer;
import javax.xml.transform.TransformerException;
import javax.xml.transform.TransformerFactory;
import javax.xml.transform.dom.DOMSource;
import javax.xml.transform.stream.StreamResult;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.log4j.Logger;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.xml.sax.SAXException;

import beans.HttpConfig;
import utils.Log4jInit;
import utils.StaticUtil;
import utils.StreamGobbler;

public class SchedByLocation {

	private static final transient Logger log = Logger.getLogger(AlgorithmThread.class);
	
	/**
	 * 根据输入文件，查询文件分块的位置，并获取节点位置的排序，指定调度任务所在节点 
	 * params: flag 是否需转换文件路径
	 * return: hostname字符串   "host1 host2..."
	 * @param fileList
	 * @return
	 */
	public static String getFileLocations(List<String> fileList,boolean flag){
		
		String fileLocations = "";
		 Configuration conf;
		 FileSystem filesystem;
		 String DEFNAME="fs.defaultFS";
			if(HttpConfig.fileSystemIP.equals("")){
						 return "";
					 }
			
		 String HDFSURL="hdfs://"+HttpConfig.fileSystemIP;
		 
		 if(flag){
			 fileList = SchedByLocation.listFormat(fileList);   
		 }
 
		 
		 Log4jInit.logger.info("-----fileLIst:"+fileList.get(0));
			
		 conf=new Configuration();
			conf.set(DEFNAME, HDFSURL);
			try {
				filesystem=FileSystem.get(conf);
				/*List<DataNodeAndLength> locList = FBSCCToWorkFlowApi.getFileLocations(filesystem, fileList);
				for(DataNodeAndLength d:locList){
					fileLocations += d.getDataNode()+" ";
				}*/
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

		return fileLocations;
	}

	/**
	 * 将windows路径转换为HDFS路径
	 * @param fileList
	 * @return
	 */
	public static List<String> listFormat(List<String> fileList) {

		List<String> rtList = new ArrayList<String>();
		if(fileList.size()==0) return null;
		
		for (String tem : fileList) {

			Log4jInit.logger.info("Before trans:" + tem);
			tem = tem.contains("\\")?tem.replace("\\", "\\\\"):tem;
			tem = tem.contains("\\\\")?tem.replace("\\\\", "/"):tem;
			String[] tems = tem.split(":/");
			tem = "/" + tems[1];
			
		Log4jInit.logger.info("FORMAT PATH:" + tem);
		
			rtList.add(tem);
		}
		return rtList;
	}

	/**
	 * 获取TASK文件中真实数据的位置
	 * @param taskFile
	 * @param flag true:根据位置调度，false禁用位置调度
	 * @param useCache:是否使用内存盘，若使用，则需修改task文件的输入输出盘符
	 * @return
	 */
public static List<String> getMdlRealPath(String taskFile,boolean flag,boolean useCache){
		
	
	//获取本地接收到的task文件
	File temFile = new File(taskFile);
	if((!temFile.exists()) || temFile.isDirectory()){
		return null;
	}
	
	//获取task文件
	 StringBuilder xmlStringBuilder = getXmlFileBuilder(temFile);
	 //获取XML文档对象
	 Document doc = stringToXml(xmlStringBuilder);
	 
	 if(useCache){
		 replaceDiskSign(doc);
			try {
				tran(doc,temFile);
			} catch (TransformerException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
	 }
	 
	 
		if(flag){
					 List<String> mdlList = new ArrayList<String>();
					 Element element = doc.getDocumentElement();
					 NodeList nl = element.getElementsByTagName("file");
					
					 for(int i=0;i<nl.getLength();i++){
						 Node n = nl.item(i);
						 Element e = (Element)n;
						 NodeList temNl = e.getElementsByTagName("value");
						 String tmp = temNl.item(0).getChildNodes().item(0).getNodeValue();  
						 Log4jInit.logger.info("mdlList path:"+tmp);
						 if(tmp.substring(tmp.lastIndexOf(".")+1).equals("mdl")){
							 mdlList.add(tmp);
						 }
						
					 }
					 
					 List<String> realPath = new ArrayList<String>();
					 
					 //转换路径，task中为O盘路径
					 mdlList = listFormat(mdlList);
					 //读取HDFS上的MDL文件，拷贝到本地，获取本地MDL路径list
					 if(mdlList != null){
						 mdlList = copyFilesFromHDFS(mdlList);
							Log4jInit.logger.info("Local Mdl:" + mdlList.get(0));
					 }else{
						 return null;
					 }
					
					 
					 //解析MDL文件
					 for(String path:mdlList){
						 File f = new File(path);
						 realPath.add(parseMdlPath(f));
						 StaticUtil.deleteFile(path);   //删除本地mdl文件
					 }
					 
					 
					 Log4jInit.logger.info("realPath path:"+realPath.get(0));
					 
				//	 List<String> hdfsPath = listFormat(realPath);
					 
					 //MDL文件里的实际路径为O盘windows路径，需转换
					return listFormat(realPath);
		}
		return null;
	}
	
	public static StringBuilder getXmlFileBuilder(File temFile){
		
		 StringBuilder xmlStringBuilder = new StringBuilder();
			
		 try {
	            BufferedReader br = new BufferedReader(new FileReader(temFile));
	            String temp = "";
	            while ((temp = br.readLine()) != null) {
	            	xmlStringBuilder.append(temp);
	            }
	            br.close();
	        } catch (FileNotFoundException e) {
	            e.printStackTrace();
	        } catch (IOException e) {
	            e.printStackTrace();
	        }
		 return xmlStringBuilder;
	}
	
	
	public static Document stringToXml(StringBuilder xmlStringBuilder){
		
		 DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
		 Document doc = null;
		 
		 try {
			InputStream is = new ByteArrayInputStream(xmlStringBuilder.toString().getBytes("utf-8"));
			doc  = dbf.newDocumentBuilder().parse(is);
			is.close();
			
		} catch (SAXException | IOException | ParserConfigurationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		return doc;
	}
	
	/**
	 * 获取MDL文件中的数据真实路径，MDL与源数据一对一
	 * @param mdlFile
	 * @return
	 */
	public static String parseMdlPath(File mdlFile){
		
		if((!mdlFile.exists())||mdlFile.isDirectory()){
			Log4jInit.logger.info("Local MDL is not exist!!");
			return null;
		}
			DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
			DocumentBuilder db;
			try {
				db = dbf.newDocumentBuilder();
				Document doc = db.parse(mdlFile);
				Element docEle = doc.getDocumentElement();
				
				 //<?xml version="1.0" encoding="UTF-8" standalone="no" ?> 也是个节点，第二个节点为真实路径值
				String realPath = docEle.getFirstChild().getNextSibling().getTextContent(); 
				Log4jInit.logger.info("realPath::"+realPath);
				//list.add(realPath);
				return realPath;
				
			} catch (ParserConfigurationException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (SAXException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		
		return null;
	}
	
	public static List<String> copyFilesFromHDFS(List<String> lists){
		
		List<String> ls = new ArrayList<String>();
		 Configuration conf;
		 FileSystem filesystem;
		 String DEFNAME="fs.defaultFS";
			
		 String HDFSURL="hdfs://"+HttpConfig.fileSystemIP;
		 if(HttpConfig.fileSystemIP.equals(""))
			 return null;
		 
		 
		 conf=new Configuration();
			conf.set(DEFNAME, HDFSURL);
			try {
				filesystem=FileSystem.get(conf);
				for(String s:lists){
					Path file = new Path(HDFSURL+s);
					FSDataInputStream in = filesystem.open(file);
					String localPath = HttpConfig.tempOutput+UUID.randomUUID().toString().substring(0,6);
					OutputStream output = new FileOutputStream(localPath);
					IOUtils.copyBytes(in, output, 4096,true);
					ls.add(localPath);
				}
				
				AlgorithmOperator.chmod(HttpConfig.tempOutput);
				
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
		return ls;
	}
	
	public static Document replaceDiskSign(Document doc){
		
		List<String> mdlList = new ArrayList<String>();
		 Element element = doc.getDocumentElement();
		 NodeList inputList = element.getElementsByTagName("file");
		 NodeList outputList = element.getElementsByTagName("AlgorithmOutput");
		
		 for(int i=0;i<inputList.getLength();i++){
			 Node n = inputList.item(i);
			 Element e = (Element)n;
			 NodeList temNl = e.getElementsByTagName("value");
			 String tmp = temNl.item(0).getChildNodes().item(0).getNodeValue();  
			 String[] str = tmp.split(":");
			 if(str.length>1){
				 tmp = HttpConfig.hdfsDisk + str[1];
				 temNl.item(0).getChildNodes().item(0).setNodeValue(tmp);
				 Log4jInit.logger.info("new input path:"+tmp);
			 }
			
		 }
		 
		 for(int i=0;i<outputList.getLength();i++){
			 Node n = outputList.item(i);
			 Element e = (Element)n;
			 NodeList temNl = e.getElementsByTagName("value");
			 String tmp = temNl.item(0).getChildNodes().item(0).getNodeValue();  
			 String[] str = tmp.split(":");
			 if(str.length>1){
				 tmp = HttpConfig.cacheDisk + str[1];
				 temNl.item(0).getChildNodes().item(0).setNodeValue(tmp);
		//		 Log4jInit.logger.info("new output path:"+tmp);
			 }
			 
		 }
		 
		
		return doc;
	}
	
	 public static void tran(Document doc,File file) throws TransformerException{
	        TransformerFactory tff = TransformerFactory.newInstance();
	        Transformer tf = tff.newTransformer();
	        tf.setOutputProperty(OutputKeys.ENCODING, "utf-8");
	        DOMSource xmlSource = new DOMSource(doc);
	        StreamResult streamresult = new StreamResult(file);
	        tf.transform(xmlSource, streamresult);
	    }
	
	
	public static String createMkdirBat(String task) {
		
		//获取本地接收到的task文件
		File temFile = new File(task);
		String  targetFilePath = HttpConfig.scriptFileUri+ UUID.randomUUID().toString().substring(0,7);
		File f = new File(targetFilePath);
		f.mkdirs();
		try {
			chmod(targetFilePath);
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
		
		 targetFilePath = targetFilePath+"/makeDir.bat";  //本地bat路径
		 Log4jInit.logger.info("makeDir.bat:"+targetFilePath);
		//获取task文件
		 StringBuilder xmlStringBuilder = getXmlFileBuilder(temFile);
		 //获取XML文档对象
		 Document doc = stringToXml(xmlStringBuilder);
		 
		
				// Set<String> mdlList = new HashSet<String>();
				 Element element = doc.getDocumentElement();
				 NodeList nl = element.getElementsByTagName("AlgorithmOutput");
				 Set<String> batSet = new HashSet<String>();
				 List<String> fileList = new ArrayList<String>();
			
				 for(int i=0;i<nl.getLength();i++){
					 Node n = nl.item(i);
					 Element e = (Element)n;
					 NodeList temNl = e.getElementsByTagName("value");
					 String tmp = temNl.item(0).getChildNodes().item(0).getNodeValue();  
					 fileList.add(tmp);
					 
					 if(tmp.indexOf(".")> 0){  //若路径中包含 .  ，则认为输入是一个文件,需处理路径，否则是一个目录，可直接创建
						 tmp.replace('/', '\\');
						 tmp = tmp.substring(0, tmp.lastIndexOf("\\"));
						 tmp.replaceAll(HttpConfig.windowsDriverLetter, HttpConfig.cacheDisk);
//						 Log4jInit.logger.info("filedir tmp::"+tmp);
						 batSet.add(tmp);
						 
					 }else{
						 //win10下，文件夹可直接move，无需对文件夹内文件进行处理
						 tmp.replaceAll(HttpConfig.windowsDriverLetter, HttpConfig.cacheDisk);
					 Log4jInit.logger.info("dir tmp::"+tmp);
					 batSet.add(tmp);
					 }
					
				 }
		
				 SchedByLocation.writeMkdirBat(targetFilePath, batSet, fileList);
		
		return targetFilePath;
	}
	
	private static File writeMkdirBat(String filePath,Set<String> bats,List<String> files){
		
		File batFile = new File(filePath);
	//	String oldFilePath = filePath.replace(HttpConfig.cacheDisk,HttpConfig.windowsDriverLetter);
		try {
			batFile.createNewFile();
			FileWriter fw = new FileWriter(batFile);
			for(String s:bats){  //mkdir command
				fw.write("mkdir ");
				fw.write(s);
				fw.write("\r\n");
			}
			
			for(String s:files){  // move file
				String tm = s.replaceAll(HttpConfig.cacheDisk, HttpConfig.windowsDriverLetter);
				fw.write("move ");
				fw.write(tm+" ");
				fw.write(s);
				fw.write("\r\n");
				
			}
			
			fw.flush();
			fw.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		return null;
	}
	
	private static void chmod(String filePath) throws IOException {

		Process p = Runtime.getRuntime().exec("chmod 777 " + filePath);

		/* 为"错误输出流"单独开一个线程读取之,否则会造成标准输出流的阻塞 */
		Thread t = new Thread(new StreamGobbler(p.getErrorStream(), "ErrorStream"));
		t.start();

		try {
			p.waitFor();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			p.destroy();
		}
	}
	
	
	public static void main(String[] args){
		//List<String> l = getMdlRealPath("/home/GF/sf/taskDemo.task",false);
		List<String> l = getMdlRealPath("/home/GF/sf/VectorChangeDetection.xml",false,true);
		
		/*for(String s:l){
			Log4jInit.logger.info("Final File Path: "+s);
		}*/
		
	}
	
	
}